package com.ds.lens.flink.monitor.common;

import com.google.common.collect.Maps;
import com.ds.lens.data.common.client.LensResponseData;
import com.ds.lens.data.common.constant.DetectionTypeEnum;
import com.ds.lens.data.common.document.DataPoint;
import com.ds.lens.data.common.document.ForecastData;
import com.ds.lens.flink.monitor.MonitorJobContext;
import com.ds.lens.flink.monitor.common.bo.Metric;
import com.ds.lens.flink.monitor.common.bo.MetricBundle;
import com.ds.lens.flink.monitor.common.bo.MonitorDefinition;
import com.ds.lens.flink.monitor.common.bo.MonitorState;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.Callable;

@Slf4j
public class MonitorDefinitionFactory {
    private LensCache<Long, MonitorDefinition> definitionCache;
    private LensCache<Long, MetricBundle> metricBundleLensCache;

    private static final MonitorDefinitionFactory INSTANCE = new MonitorDefinitionFactory();

    private MonitorDefinitionFactory() {
        int cacheSize = MonitorJobContext.getInstance().getMaxCacheSize();
        int expired = MonitorJobContext.getInstance().getCacheExpiredInMinute();
        definitionCache = new LensCache<>(new MonitorDefinitionLoadFunction(), cacheSize, expired);
        metricBundleLensCache = new LensCache<>(null, 10000, 30);
    }

    public static MonitorDefinitionFactory get() {
        return INSTANCE;
    }

    public MonitorDefinition getMonitorDefinitionByState(MonitorState monitorState) {
        Long monitorId = monitorState.getKey().getMonitorId();
        if (monitorState.getIsDelete()) {
            definitionCache.invalidate(monitorId);
            return null;
        }
        MonitorDefinition definition = definitionCache.get(monitorId);
        if (definition != null &&
                (!Objects.equals(monitorState.getKey().getVersion(), definition.getMonitorState().getKey().getVersion()) ||
                        !monitorState.getModifyTime().equals(definition.getMonitorState().getModifyTime()))) {
            definition = definitionCache.reload(monitorId);
        }
        if (definition != null) {
            definition.getMonitorState().setApplicationName(monitorState.getApplicationName());
        }
        return definition;
    }

    public MetricBundle getMetricBundleByState(MonitorState state, long startTime, long endTime, long interval) {
        MonitorDefinition definition = getMonitorDefinitionByState(state);
        state = definition.getMonitorState();
        MetricBundleCallable bundleCallable = new MetricBundleCallable(definition, startTime, endTime, interval);
        MetricBundle bundle = metricBundleLensCache.get(state.getKey().getMonitorId(), bundleCallable);
        LogUtils.debug("query metrics: state={}, startTime={}, endTime={}, interval={}, bundle={}",
                state, startTime, endTime, interval, bundle);
        if (bundle != null) {
            if (endTime > bundle.getEndTime()) {
                if (!bundle.getState().getKey().getMonitorId().equals(state.getKey().getMonitorId()) ||
                        !bundle.getState().getKey().getVersion().equals(state.getKey().getVersion()) ||
                        startTime >= bundle.getEndTime()) {
                    bundle = metricBundleLensCache.reload(state.getKey().getMonitorId(), bundleCallable);
                    LogUtils.debug("reload metrics: state={}, startTime={}, endTime={}, interval={}, bundle={}",
                            state, startTime, endTime, interval, bundle);
                } else {
                    MetricBundle newBundle = retriveMetricBundle(definition, bundle.getEndTime(), endTime, interval);
                    LogUtils.debug("append metrics: state={}, startTime={}, endTime={}, interval={}, bundle={}",
                            state, startTime, endTime, interval, newBundle);
                    bundle = mergeMetricBundle(state, newBundle, bundle, startTime, endTime);
                    if (bundle != null) {
                        metricBundleLensCache.put(definition.getMonitorState().getKey().getMonitorId(), bundle);
                    } else {
                        metricBundleLensCache.invalidate(definition.getMonitorState().getKey().getMonitorId());
                    }
                }
            }
        }

        if (bundle == null) {
            bundle = BuilderUtil.buildMetricBundle(definition, null, null, startTime, endTime);
            LogUtils.debug("empty metrics: state={}, startTime={}, endTime={}, interval={}, bundle={}",
                    state, startTime, endTime, interval, bundle);
        }
        LogUtils.debug("final metrics: state={}, startTime={}, endTime={}, interval={}, bundle={}",
                state, startTime, endTime, interval, bundle);
        return bundle;
    }

    public void reload(MonitorState monitorState) {
        definitionCache.reload(monitorState.getKey().getMonitorId());
    }

    private MetricBundle retriveMetricBundle(MonitorDefinition definition, long startTime, long endTime, long interval) {
        LensResponseData<List<Map<String, Object>>> resp = MonitorJobContext.getInstance()
                .getMonitorClient().queryMetrics(definition.getMonitorState().getKey().getMonitorId(), startTime, endTime, interval);

        Map<String, ForecastData> forecastDataMap = null;

        if (definition.getDetectionMethod() == DetectionTypeEnum.DYNAMIC_STATE.getCode()) {
            LensResponseData<List<ForecastData>> forecastDatas = MonitorJobContext.getInstance().getMonitorClient()
                    .queryForecastMetrics(definition.getMonitorState().getKey().getMonitorId(), definition.getMonitorState().getKey().getVersion(),
                            startTime, endTime);

            if (!CollectionUtils.isEmpty(forecastDatas.getData())) {
                forecastDataMap = Maps.newHashMapWithExpectedSize(forecastDatas.getData().size());
                for (ForecastData forecastData : forecastDatas.getData()) {
                    forecastDataMap.put(forecastData.getId(), forecastData);
                }
            }
        }
        List<Map<String, Object>> list = resp.getData();
        if (!CollectionUtils.isEmpty(list)) {
            MetricBundle bundle = BuilderUtil.buildMetricBundle(definition, forecastDataMap, list, startTime, endTime);
            LogUtils.debug("retriveMetricBundle request monitorId={}, startTime={}, endTime={} interval={}, resp={}, metricBundle={}",
                    definition.getMonitorState().getKey().getMonitorId(), startTime, endTime, interval, list, bundle);
            return bundle;
        } else {
            LogUtils.debug("retriveMetricBundle request monitorId={}, startTime={}, endTime={} interval={}, resp={}, metricBundle={}",
                    definition.getMonitorState().getKey().getMonitorId(), startTime, endTime, interval, list, null);
            return null;
        }
    }

    public MetricBundle mergeMetricBundle(MonitorState state, MetricBundle newBundle, MetricBundle oldBundle, long startTime, long endTime) {
        Map<String, Metric> newMetricMap = (newBundle == null || CollectionUtils.isEmpty(newBundle.getMetrics())) ?
                new HashMap<>() : newBundle.getMetrics();
        Map<String, Metric> oldMetricMap = (oldBundle == null || CollectionUtils.isEmpty(oldBundle.getMetrics())) ?
                new HashMap<>() : oldBundle.getMetrics();
        for (Iterator<Map.Entry<String, Metric>> entryIterator = oldMetricMap.entrySet().iterator(); entryIterator.hasNext(); ) {
            Map.Entry<String, Metric> entry = entryIterator.next();
            String key = entry.getKey();
            Metric oldMetric = entry.getValue();
            List<DataPoint> oldDataPoints = oldMetric.getValues();
            if (!CollectionUtils.isEmpty(oldDataPoints)) {
                for (Iterator<DataPoint> itr = oldDataPoints.iterator(); itr.hasNext(); ) {
                    DataPoint dataPoint = itr.next();
                    if (dataPoint.getTimeStamp() < startTime) {
                        itr.remove();
                    }
                }
            }

            if (!CollectionUtils.isEmpty(newMetricMap) && newMetricMap.containsKey(key)) {
                Metric newMetric = newMetricMap.remove(key);
                for (DataPoint dataPoint : newMetric.getValues()) {
                    int index = oldDataPoints.indexOf(dataPoint);
                    if (index >= 0) {
                        oldDataPoints.set(index, dataPoint);
                    } else {
                        oldDataPoints.add(dataPoint);
                    }
                }
            }

            if (oldDataPoints.isEmpty()) {
                entryIterator.remove();
            }
        }

        if (!CollectionUtils.isEmpty(newMetricMap)) {
            for (String key : newMetricMap.keySet()) {
                Metric newMetric = newMetricMap.get(key);
                oldMetricMap.put(key, newMetric);
            }
        }

        if (!CollectionUtils.isEmpty(oldMetricMap)) {
            for (String key : oldMetricMap.keySet()) {
                Metric metric = oldMetricMap.get(key);
                metric.setStartTime(startTime);
                metric.setEndTime(endTime);
            }
        }

        if (!CollectionUtils.isEmpty(oldMetricMap)) {
            MetricBundle mergedBundle = new MetricBundle(state);
            mergedBundle.setStartTime(startTime);
            mergedBundle.setEndTime(endTime);
            mergedBundle.setMetrics(oldMetricMap);
            LogUtils.debug("merged metric bundle={}, from new bundle uid = {} and old bundle uid = {}", mergedBundle, newBundle, oldBundle);
            return mergedBundle;
        } else {
            LogUtils.debug("merged metric bundle={}, from new bundle uid = {} and old bundle uid = {}", null, newBundle, oldBundle);
            return null;
        }
    }

    private class MetricBundleCallable implements Callable<MetricBundle> {
        private MonitorDefinition definition;
        private long startTime;
        private long endTime;
        private long interval;

        public MetricBundleCallable(MonitorDefinition definition, long startTime, long endTime, long interval) {
            this.definition = definition;
            this.startTime = startTime;
            this.endTime = endTime;
            this.interval = interval;
        }

        @Override
        public MetricBundle call() throws Exception {
            MetricBundle bundle = retriveMetricBundle(definition, startTime, endTime, interval);
            if (bundle == null) {
                throw new ObjectNotFoundException();
            }
            return bundle;
        }
    }
}
